package com.jdrx.sw.sewage.task.xxjob;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.jdrx.platform.commons.rest.exception.BizException;
import com.jdrx.sw.sewage.beans.entity.*;
import com.jdrx.sw.sewage.beans.vo.GatherDevPointOptionVO;
import com.jdrx.sw.sewage.beans.vo.SimpleRealtimeDataVO;
import com.jdrx.sw.sewage.common.enums.ReportType;
import com.jdrx.sw.sewage.common.helper.HttpClientHelper;
import com.jdrx.sw.sewage.common.util.Asserts;
import com.jdrx.sw.sewage.common.util.LocalDateTimeUtils;
import com.jdrx.sw.sewage.dao.*;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.handler.annotation.XxlJob;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.math.NumberUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.function.Function;
import java.util.stream.Collectors;

import static com.jdrx.sw.sewage.common.constant.Constant.DATA_ANALYSIS_12HOUR;
import static com.jdrx.sw.sewage.common.constant.Constant.DATA_ANALYSIS_1HOUR;

@Component
public class SewageJob {
    private static final Logger logger = LoggerFactory.getLogger(SewageJob.class);
    @Value("${spring.application.name}")
    private String platformCode;
    @Value("${forecast.pumpChokeUrl}")
    private String pumpChokeUrl;
    @Autowired
    private DeviceTimeDataDAO deviceTimeDataDAO;
    @Autowired
    private GatherDevPointDAO gatherDevPointDAO;
    @Autowired
    private StationReportAnalysisDAO stationReportAnalysisDAO;
    @Autowired
    private CdForecastDataDAO cdForecastDataDAO;
    @Autowired
    private CdRealtimeDataDAO cdRealtimeDataDAO;
    @Autowired
    private CdAiLogDAO cdAiLogDAO;
    @Autowired
    private AsyncSewageJob asyncSewageJob;
    @Autowired
    private AiModelDAO aiModelDAO;

    /**
     * 每晚零点定时初始化告警编号初始值为1
     */
    @XxlJob("alarmNumInitValueJob")
    public ReturnT<String> alarmNumInitValueJob(String dataNodes) throws Exception {
        //null 为一体机
        if (StringUtils.isEmpty(dataNodes)) {
            asyncSewageJob.alarmNumInitValueJob2(null);
        } else {
            String[] dataNodeArr = dataNodes.split(",");
            if (dataNodeArr.length > 0) {
                for (int j = 0; j < dataNodeArr.length; j++) {
                    String dataNode = dataNodeArr[j];
                    dataNode = getSQLPerfix(dataNode);
                    asyncSewageJob.alarmNumInitValueJob2(dataNode);
                }
            }
        }
        return ReturnT.SUCCESS;
    }


    /**
     * 清晰数据分析-日报表- 每天整点
     */
    @XxlJob("deviceTimeDataHourFilterJob")
    public ReturnT<String> deviceTimeDataHourFilterJob(String dataNodes) throws Exception {
        Integer reportType = ReportType.DAY_REPORT.getKey();
        LocalDateTime now = LocalDateTime.now();
        if (StringUtils.isEmpty(dataNodes))
            asyncSewageJob.deviceTimeDataHourFilterJob2(reportType, now, null, DATA_ANALYSIS_1HOUR, LocalDateTimeUtils.reduceHour(1));
        else {
            String[] dataNodeArr = dataNodes.split(",");
            if (dataNodeArr.length > 0) {
                for (int i = 0; i < dataNodeArr.length; i++) {
                    String dataNode = dataNodeArr[i];
                    dataNode = getSQLPerfix(dataNode);
                    asyncSewageJob.deviceTimeDataHourFilterJob2(reportType, now, dataNode, DATA_ANALYSIS_1HOUR, LocalDateTimeUtils.reduceHour(1));
                }
            }
        }
        return ReturnT.SUCCESS;
    }

    /**
     * 清晰数据分析-月和年报表- 每天0点和12点
     */
    @XxlJob("dataAnalysisMonthAndYearJob")
    public ReturnT<String> dataAnalysisMonthAndYearJob(String dataNodes) throws Exception {
        Integer reportType = ReportType.MONTH_REPORT.getKey();
        LocalDateTime now = LocalDateTime.now();
        if (StringUtils.isEmpty(dataNodes))
            asyncSewageJob.deviceTimeDataHourFilterJob2(reportType, now, null, DATA_ANALYSIS_12HOUR, LocalDateTimeUtils.reduceHour(12));
        else {
            String[] dataNodeArr = dataNodes.split(",");
            if (dataNodeArr.length > 0) {
                for (int j = 0; j < dataNodeArr.length; j++) {
                    String dataNode = dataNodeArr[j];
                    dataNode = getSQLPerfix(dataNode);
                    asyncSewageJob.deviceTimeDataHourFilterJob2(reportType, now, dataNode, DATA_ANALYSIS_12HOUR, LocalDateTimeUtils.reduceHour(12));
                }
            }
        }
        return ReturnT.SUCCESS;
    }

    /**
     * 碳源加药间1#流量监测
     * @param param
     * @return
     */
    @XxlJob("tyjy1MonitorJob")
    @Transactional
    public ReturnT<String> tyjy1Monitor(String param) throws Exception {
    	logger.info("碳源加药间1#流量监测 - 执行开始，执行时间{}", LocalDateTime.now());
    	
    	DateTimeFormatter dtf = DateTimeFormatter.ofPattern("yyyy-MM-dd HH");
    	
    	LocalDateTime now = LocalDateTime.now();
    	String time = dtf.format(now);
    	List<SimpleRealtimeDataVO> list = deviceTimeDataDAO.findTyjy1ll(time);
    	list = list.stream().filter(p -> NumberUtils.isCreatable(p.getValue())).collect(Collectors.toList());
    	logger.debug("当前日期{}1#碳源加药间瞬时流量数据条数{}", now, list.size());
    	if(CollectionUtils.isNotEmpty(list)) {
    		double sum = list.stream().mapToDouble(p -> Double.parseDouble(p.getValue())).sum();
    		double ll = new BigDecimal(sum).divide(new BigDecimal(list.size()), BigDecimal.ROUND_HALF_UP).setScale(3, BigDecimal.ROUND_DOWN).doubleValue();
    	
    		// 获取当时预测的值
    		CdForecastDataPO po = cdForecastDataDAO.findByTime("碳源", time);
    		String value = Optional.ofNullable(po.getValue()).orElse("0");
    		double doubleValue = new BigDecimal(value).setScale(2, BigDecimal.ROUND_DOWN).doubleValue();
    		
    		logger.debug("当前日期{}统计的1#碳源加药间瞬时流量{},预测流量{}", ll, doubleValue);
    		// 写入AI对话日志
    		CdAiLogPO log = new CdAiLogPO();
    		log.setCreateAt(LocalDateTime.now());
    		if(ll < doubleValue) {
    			log.setLevel(0);
    			log.setEvent("实际投加药剂量" + ll + "L,比预测值" + doubleValue + "L少,建议启动对应的计量泵");
    		} else {
    			log.setLevel(1);
    			log.setEvent("实际投加药剂量" + ll + "L,比预测值" + doubleValue + "L多,建议停止对应的计量泵");
    		}
    		cdAiLogDAO.insert(log);
    	}
    	
    	logger.info("碳源加药间1#流量监测 - 执行完成，执行时间{}", LocalDateTime.now());
    	return ReturnT.SUCCESS;
    }
    
    /**
     * 碳源加药间1#流量按小时清洗
     * @return
     */
    @XxlJob("tyjy1llJob")
    @Transactional
    public ReturnT<String> tyjy1ll(String param) throws Exception {
    	logger.info("按小时清洗一厂碳源加药间1#流量 - 执行开始，执行时间{}", LocalDateTime.now());
    	DateTimeFormatter dtf = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    	// 清洗上1个小时的数据
    	LocalDateTime lastHour = LocalDateTime.now().minusHours(1);
    	List<SimpleRealtimeDataVO> list = deviceTimeDataDAO.findTyjy1ll(DateTimeFormatter.ofPattern("yyyy-MM-dd HH").format(lastHour));
    	list = list.stream().filter(p -> NumberUtils.isCreatable(p.getValue())).collect(Collectors.toList());
    	if(CollectionUtils.isNotEmpty(list)) {
    		// 小时累计流量=瞬时流量和/数据条数
    		double sum = list.stream().mapToDouble(p -> Double.parseDouble(p.getValue())).sum();
    		double ll = new BigDecimal(sum).divide(new BigDecimal(list.size()), BigDecimal.ROUND_HALF_UP).setScale(3, BigDecimal.ROUND_DOWN).doubleValue();
    		CdRealtimeDataPO po = new CdRealtimeDataPO();
    		po.setName("碳源");
    		po.setUnit("L");
    		po.setDataTime(LocalDateTime.parse(dtf.format(lastHour.withMinute(0).withSecond(0)), dtf));
    		po.setValue(String.valueOf(ll));
    		cdRealtimeDataDAO.insert(po);
    		logger.info("按小时清洗一厂碳源加药间1#流量 - 数据插入成功{}，执行时间{}", ll, LocalDateTime.now());
    	}
    	logger.info("按小时清洗一厂碳源加药间1#流量 - 执行完成，执行时间{}", LocalDateTime.now());
    	return ReturnT.SUCCESS;
    }

    /**
     * 碳源加药预测
     *
     * @param
     * @return
     * @throws Exception
     */
    @XxlJob("carbonDrugJob")
    public ReturnT<String> carbonDrug(String url) throws Exception {
    	logger.info("智能碳源加药 - 执行开始，执行时间{}", LocalDateTime.now());
        // 预测1小时后的加药量
    	LocalDateTime now = LocalDateTime.now().withMinute(0);
        String predictTime = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm").format(now.plusHours(1));
        String lastTime = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm").format(now.minusHours(1));
        // 模型入参
        // 进水口实时数据
     	List<GatherDevPointOptionVO> jsk = gatherDevPointDAO.findByValues(Lists.newArrayList("JSK_NH3NY_NH3N_Y", "JSK_PHY_PH_Y", "JSK_SSY_SS_Y", "JSK_TNY_TN_Y", "JSK_TPY_TP_Y", "JSK_WDJ_WD_Y", "JSK_CODY_COD_Y", "JSK_LLJ_RJSLL_Y"));
     	// 按小时后清洗的进水累计流量
     	List<StationReportAnalysisPO> ljll = stationReportAnalysisDAO.findByCondition(5, 1, lastTime);
     	// 一厂一期生化池实时数据
     	List<GatherDevPointOptionVO> shc = gatherDevPointDAO.findByValues(Lists.newArrayList("SHC1_DOY1_DO_Y", "SHC1_DOY2_DO_Y", "SHC1_MLSSY1_MLSS_Y", "SHC1_MLSSY2_MLSS_Y", "SHC1_ORPY1_ORP_Y", "SHC1_ORPY2_ORP_Y"));
        
     	/*
     	 * 根据实际情况来说，每个污水厂的情况不同，且模型主要通过该水厂的历史数据训练而来，所以模型不存在通用。
     	 * 就算一个模型，对应不同的水厂，也要通过不同的数据训练，执行的预测结果也会大相径庭，所以正确的做法应该
     	 * 是开发一个模型管理工具，针对不同的租户污水厂设置不同的模型地址及参数，这里仅仅针对金海污水厂起作用，
     	 * 后续会开发一个模型管理工具
     	 */
     	Map<String, GatherDevPointOptionVO> jskMap = jsk.stream().collect(Collectors.toMap(GatherDevPointOptionVO :: getCode, Function.identity(), (key1, key2) -> key2));
     	Map<String, GatherDevPointOptionVO> shcMap = shc.stream().collect(Collectors.toMap(GatherDevPointOptionVO :: getCode, Function.identity(), (key1, key2) -> key2));
     	
     	/*组装进水口参数*/
     	Map<String, Object> predictInMap = Maps.newHashMap();
     	// 前1个小时进水累计流量
     	if(ljll.size() == 0) {
     		predictInMap.put("flow", null);
     	} else {
            predictInMap.put("flow", Optional.ofNullable(ljll.get(0).getLjll()).orElse(null));
        }
     	// 温度
     	predictInMap.put("temp", Optional.ofNullable(jskMap.get("JSK_WDJ_WD_Y").getValue()).orElse(null));
     	predictInMap.put("nh3", Optional.ofNullable(jskMap.get("JSK_NH3NY_NH3N_Y").getValue()).orElse(null));
     	predictInMap.put("tp", Optional.ofNullable(jskMap.get("JSK_TPY_TP_Y").getValue()).orElse(null));
     	predictInMap.put("tn", Optional.ofNullable(jskMap.get("JSK_TNY_TN_Y").getValue()).orElse(null));
     	predictInMap.put("ss", Optional.ofNullable(jskMap.get("JSK_SSY_SS_Y").getValue()).orElse(null));
     	predictInMap.put("ph", Optional.ofNullable(jskMap.get("JSK_PHY_PH_Y").getValue()).orElse(null));
     	predictInMap.put("cod", Optional.ofNullable(jskMap.get("JSK_CODY_COD_Y").getValue()).orElse(null));
     	/*组装生化池参数*/
     	Map<String, Object> predictShcMap = Maps.newHashMap();
     	predictShcMap.put("mlss1", Optional.ofNullable(shcMap.get("SHC1_MLSSY1_MLSS_Y").getValue()).orElse(null));
     	predictShcMap.put("mlss2", Optional.ofNullable(shcMap.get("SHC1_MLSSY2_MLSS_Y").getValue()).orElse(null));
     	predictShcMap.put("orp1", Optional.ofNullable(shcMap.get("SHC1_ORPY1_ORP_Y").getValue()).orElse(null));
     	predictShcMap.put("orp2", Optional.ofNullable(shcMap.get("SHC1_ORPY2_ORP_Y").getValue()).orElse(null));
     	/*组装最终参数*/
     	Map<String, Object> predictMap = Maps.newHashMap();
     	predictMap.put("in", predictInMap);
     	predictMap.put("pool", predictShcMap);
     	predictMap.put("time", predictTime);
     	
     	// 参数校验
     	for(Map.Entry<String, Object> entry : predictInMap.entrySet()) {
     		if(entry.getValue() == null) {
     			logger.info("智能碳源加药 - 调用模型执行参数{}缺失，执行时间{}", entry.getKey(), LocalDateTime.now());
     			return ReturnT.SUCCESS;
     		}
     	}
     	for(Map.Entry<String, Object> entry : predictShcMap.entrySet()) {
     		if(entry.getValue() == null) {
     			logger.info("智能碳源加药 - 调用模型执行参数{}缺失，执行时间{}", entry.getKey(), LocalDateTime.now());
     			return ReturnT.SUCCESS;
     		}
     	}
     	
     	// 调用模型
     	String result = HttpClientHelper.sendPost(url, JSON.toJSONString(predictMap));
     	logger.info("智能碳源加药 - 调用模型执行完成，结果返回为{}，执行时间{}", result, LocalDateTime.now());
     	if (StringUtils.isNotEmpty(result)) {
     		DateTimeFormatter dtf = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
     		result = NumberUtils.isCreatable(result) ? new BigDecimal(result).setScale(2, BigDecimal.ROUND_DOWN).toString() : "";

     		CdForecastDataPO po = new CdForecastDataPO();
     		po.setName("碳源");
     		po.setValue(result);
     		po.setUnit("L");
     		po.setDataTime(LocalDateTime.parse(dtf.format(now.plusHours(1).withMinute(0).withSecond(0)), dtf));
     		cdForecastDataDAO.insert(po);
     		logger.info("智能碳源加药 - 执行完成，数据添加至数据库，执行时间{}", LocalDateTime.now());
     		
     		// 写入AI对话日志
    		CdAiLogPO log = new CdAiLogPO();
    		log.setCreateAt(LocalDateTime.now());
    		log.setLevel(0);
			log.setEvent("预测未来" + now.plusHours(1).getHour() + ":00-" + now.plusHours(2).getHour() + ":00需要投加" +result+ "L药剂");
    		cdAiLogDAO.insert(log);
     	}
     	logger.info("智能碳源加药 - 执行完成，执行时间{}", LocalDateTime.now());
     	return ReturnT.SUCCESS;
    }

    /**
     * 运营简报推送
     */
    @XxlJob("operateBriefPush")
    public ReturnT<String> operateBriefPush(String dataNodes) throws Exception {
        if (StringUtils.isEmpty(dataNodes)) asyncSewageJob.operateBriefPush2(null);
        else {
            String[] dataNodeArr = dataNodes.split(",");
            if (dataNodeArr.length > 0) {
                for (int j = 0; j < dataNodeArr.length; j++) {
                    String dataNode = dataNodeArr[j];
                    dataNode = getSQLPerfix(dataNode);
                    asyncSewageJob.operateBriefPush2(dataNode);
                }
            }
        }
        return ReturnT.SUCCESS;
    }

    /**
     * 每天零点执行维保自然日自动减一天
     */
    @XxlJob("maintenanceCycleDay")
    private ReturnT<String> maintenanceCycleDay(String dataNodes) throws BizException {
        if (StringUtils.isEmpty(dataNodes)) asyncSewageJob.maintenanceCycleDay2(null);
        else {
            String[] dataNodeArr = dataNodes.split(",");
            if (dataNodeArr.length > 0) {
                for (int j = 0; j < dataNodeArr.length; j++) {
                    String dataNode = dataNodeArr[j];
                    dataNode = getSQLPerfix(dataNode);
                    asyncSewageJob.maintenanceCycleDay2(dataNode);
                }
            }
        }
        return ReturnT.SUCCESS;
    }

    /**
     * 每整小时执行一次
     */
    @XxlJob("maintenanceCycleHour")
    private ReturnT<String> maintenanceCycleHour(String dataNodes) throws BizException {
        if (StringUtils.isEmpty(dataNodes)) asyncSewageJob.maintenanceCycleHour2(null);
        else {
            String[] dataNodeArr = dataNodes.split(",");
            if (dataNodeArr.length > 0) {
                for (int j = 0; j < dataNodeArr.length; j++) {
                    String dataNode = dataNodeArr[j];
                    dataNode = getSQLPerfix(dataNode);
                    asyncSewageJob.maintenanceCycleHour2(dataNode);
                }
            }
        }
        return ReturnT.SUCCESS;
    }

    /**
     * 数据大屏水质监测拉-整点取数据
     */
    @XxlJob("stationWaterQuality")
    public ReturnT<String> stationWaterQuality(String dataNodes) throws BizException {
        if (StringUtils.isEmpty(dataNodes)) asyncSewageJob.stationWaterQuality2(null);
        else {
            String[] dataNodeArr = dataNodes.split(",");
            if (dataNodeArr.length > 0) {
                for (int j = 0; j < dataNodeArr.length; j++) {
                    String dataNode = dataNodeArr[j];
                    dataNode = getSQLPerfix(dataNode);
                    asyncSewageJob.stationWaterQuality2(dataNode);
                }
            }
        }
        return ReturnT.SUCCESS;
    }

    /**
     * 实时数据表备份-暂定1个小时执行一次
     */
    @XxlJob("deviceTimeDataBackupJob")
    public ReturnT<String> deviceTimeDataBackupJob(String dataNodes) throws BizException {
        LocalDateTime now = LocalDateTime.now();
        if (StringUtils.isEmpty(dataNodes)) asyncSewageJob.deviceTimeDataBackup2(null, now);
        else {
            String[] dataNodeArr = dataNodes.split(",");
            if (dataNodeArr.length > 0) {
                for (int j = 0; j < dataNodeArr.length; j++) {
                    String dataNode = dataNodeArr[j];
                    dataNode = getSQLPerfix(dataNode);
                    asyncSewageJob.deviceTimeDataBackup2(dataNode, now);
                }
            }
        }
        return ReturnT.SUCCESS;
    }

    /**
     * ai模型执行
     */
    @XxlJob("aiModelJob")
    public ReturnT<String> aiModelJob(String corn) throws BizException {
        if (StringUtils.isEmpty(corn)) return ReturnT.FAIL;
        else aiModelJob2(corn);
        return ReturnT.SUCCESS;
    }

    /**
     * AI模型周报生成(每周一早上8.30生成并推送微信)
     */
    @XxlJob("aiModeWeekReportJob")
    public ReturnT<String> aiModeWeekReportJob(String dataNodes) throws BizException {
        if (StringUtils.isEmpty(dataNodes)) asyncSewageJob.aiModeWeekReportJob(null);
        else {
            String[] dataNodeArr = dataNodes.split(",");
            if (dataNodeArr.length > 0) {
                for (int j = 0; j < dataNodeArr.length; j++) {
                    String dataNode = dataNodeArr[j];
                    dataNode = getSQLPerfix(dataNode);
                    asyncSewageJob.aiModeWeekReportJob(dataNode);
                }
            }
        }
        return ReturnT.SUCCESS;
    }

    public void aiModelJob2(String corn) {
        List<AiModelPO> aiModelPOList = aiModelDAO.findAiModelByCron(corn);
        if (null == aiModelPOList && aiModelPOList.size() <= 0) {
            logger.info("ai模型执行 - 模型配置错误或为空，不处理ai调用{}", LocalDateTime.now());
            return;
        }

        Date now = new Date();
        LocalDateTime ymdh = LocalDateTimeUtils.stringToLocalDateTime(LocalDateTimeUtils.dateCURRToString2(now));
        aiModelPOList.forEach(item -> {
            asyncSewageJob.aiModelExecute(item, ymdh, now);
            asyncSewageJob.aiReport(item, ymdh, now);
        });
    }

    protected String getSQLPerfix(String dataNode) throws BizException {
        Asserts.notBlank(dataNode, "租户id");
        return "/*!mycat:schema = " + dataNode + "*/";
    }
}